package client

import (
	"context"
	"errors"
	"fmt"
	"time"

	logger "github.com/sirupsen/logrus"
	"go.etcd.io/etcd/clientv3"
)

type Etcd struct {
	endpoints []string
	client    *clientv3.Client
	kv        clientv3.KV
	timeout   time.Duration
}

func NewEtcd(endpoints []string, timeout time.Duration) (*Etcd, error) {
	client, err := clientv3.New(clientv3.Config{
		Endpoints:   endpoints,
		DialTimeout: timeout,
	})

	if err != nil {
		return nil, err
	}

	return &Etcd{
		endpoints: endpoints,
		client:    client,
		kv:        clientv3.NewKV(client),
		timeout:   timeout,
	}, nil
}

func (etcd *Etcd) Get(key string) ([]byte, error) {

	ctx, cancelFunc := context.WithTimeout(context.Background(), etcd.timeout)
	defer cancelFunc()
	getResponse, err := etcd.kv.Get(ctx, key)

	if err != nil {
		return nil, err
	}

	if len(getResponse.Kvs) == 0 {
		return nil, errors.New("key[ " + key + " ] does not exist")
	}

	return getResponse.Kvs[0].Value, nil
}

func (etcd *Etcd) GetWithPrefixLimit(keyPrefix string, limit int64) ([][]byte, [][]byte, error) {
	ctx, cancelFunc := context.WithTimeout(context.Background(), etcd.timeout)
	defer cancelFunc()

	getResp, err := etcd.client.Get(ctx, keyPrefix, clientv3.WithPrefix(), clientv3.WithLimit(limit))

	if err != nil {
		return nil, nil, err
	}

	if len(getResp.Kvs) == 0 {
		return nil, nil, errors.New(fmt.Sprintf("got nothing from the key prefix: %s", keyPrefix))
	}

	keys, values := make([][]byte, 0), make([][]byte, 0)
	for _, kv := range getResp.Kvs {
		keys = append(keys, kv.Key)
		values = append(values, kv.Value)
	}

	return keys, values, nil
}

func (etcd *Etcd) Put(key, value string) error {
	ctx, cancelFunc := context.WithTimeout(context.Background(), etcd.timeout)
	defer cancelFunc()

	if _, err := etcd.kv.Put(ctx, key, value); err != nil {
		return err
	}
	return nil
}

func (etcd *Etcd) TxKeepaliveWithTTL(key, value string, ttl int64) (*TxResponse, error) {

	lease := clientv3.NewLease(etcd.client)
	leaseGrantResp, err := lease.Grant(context.Background(), ttl)
	if err != nil {
		return nil, err
	}
	leaseID := leaseGrantResp.ID

	keepAliveResp, err := lease.KeepAlive(context.Background(), leaseID)
	if err != nil {
		return nil, err
	}

	txResponse := &TxResponse{
		LeaseID: leaseID,
		Lease:   lease,
	}

	go func() {
		for ch := range keepAliveResp {
			if ch == nil {
				logger.Warnf("the tx keepalive has lose key: %s", key)
				if txResponse.StateChan != nil {
					txResponse.StateChan <- false
				}
				break
			}
		}
	}()

	ctx, cancelFunc := context.WithTimeout(context.Background(), etcd.timeout)
	defer cancelFunc()

	txn := etcd.client.Txn(ctx)

	txnResponse, err := txn.
		If(clientv3.Compare(clientv3.Version(key), "=", 0)).
		Then(clientv3.OpPut(key, value, clientv3.WithLease(leaseID))).
		Else(clientv3.OpGet(key)).
		Commit()

	if err != nil {
		_ = lease.Close()
		return nil, err
	}

	if txnResponse.Succeeded {
		txResponse.Success = true
		txResponse.StateChan = make(chan bool, 0)
	} else {
		// close the lease
		_ = lease.Close()
		txResponse.Success = false
		v, err := etcd.Get(key)
		if err != nil {
			return nil, err
		}
		txResponse.Key = key
		txResponse.Value = string(v)
	}

	return txResponse, nil
}

func (etcd *Etcd) Delete(key string) error {
	ctx, cancelFunc := context.WithTimeout(context.Background(), etcd.timeout)
	defer cancelFunc()

	_, err := etcd.client.Delete(ctx, key)
	return err
}
